package com.sc.passbook.service.impl;

import com.google.common.collect.Lists;
import com.sc.passbook.constant.Constant;
import com.sc.passbook.service.IUserService;
import com.sc.passbook.vo.Response;
import com.sc.passbook.vo.User;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @author sc
 * Created on  2018/9/1
 */
@Service
@Slf4j
public class UserServiceImpl implements IUserService {

    @Autowired
    private HbaseTemplate hbaseTemplate;
    @Autowired
    private StringRedisTemplate redisTemplate;


    @Override
    public Response createUser(User user) {
        byte[] FAMILY_B = Constant.UserTable.FAMILY_B.getBytes();
        byte[] NAME = Constant.UserTable.NAME.getBytes();
        byte[] AGE = Constant.UserTable.AGE.getBytes();
        byte[] GENDER = Constant.UserTable.GENDER.getBytes();

        byte[] FAMILY_O = Constant.UserTable.FAMILY_O.getBytes();
        byte[] PHONE = Constant.UserTable.PHONE.getBytes();
        byte[] ADDRESS = Constant.UserTable.ADDRESS.getBytes();

        Long count = redisTemplate.opsForValue().increment(Constant.USE_COUNT_REDIS_KEY, 1);
        Long userId = getUserId(count);
        List<Mutation> data  = Lists.newArrayList();
        Put put = new Put(Bytes.toBytes(userId));
        put.addColumn(FAMILY_B,NAME,Bytes.toBytes(user.getBaseInfo().getName()));
        put.addColumn(FAMILY_B,AGE,Bytes.toBytes(user.getBaseInfo().getAge()));
        put.addColumn(FAMILY_B,GENDER,Bytes.toBytes(user.getBaseInfo().getGender()));
        put.addColumn(FAMILY_O,PHONE,Bytes.toBytes(user.getOtherInfo().getPhone()));
        put.addColumn(FAMILY_O,ADDRESS,Bytes.toBytes(user.getOtherInfo().getAddres()));
        data.add(put);
        hbaseTemplate.saveOrUpdates(Constant.UserTable.TABLE_NAME,data);
        user.setId(userId);
        return Response.success(user);
    }

    private Long getUserId(Long prefix) {
        return Long.valueOf(prefix + RandomStringUtils.randomNumeric(5));
    }
}
